package org.example.netty.handler.channel;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.example.netty.constant.MessageType;
import org.example.netty.constant.ProtocolConstants;
import org.example.netty.mgt.ConnectionManager;
import org.example.netty.model.ExtendedChannelContext;
import org.example.netty.model.CustomMessage;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class HeartbeatHandler extends AbstractBusinessHandler {
    private static final int MAX_TIMEOUT_RETRY = 3;      // 最大连续超时次数
    private static final long HEARTBEAT_INTERVAL_MS = 300_000;  // 5分钟
    private static final long HEARTBEAT_TIMEOUT_MS = 5_000;     // 5秒超时
    private ScheduledFuture<?> scheduledTask;
    private final AtomicInteger timeoutCount = new AtomicInteger(0);
    /* 共享线程池（按CPU核心数配置）*/
    private static final ScheduledExecutorService GLOBAL_SCHEDULER =
            Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors());

    public HeartbeatHandler(ConnectionManager connectionManager) {
        super(connectionManager);

    }

    /**
     * 重写 channelActive 当第一次连接时 将，
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //将 channel 放入 connectionManager中进行管理
        if (connectionManager.getChannel(ctx.channel().id())==null){//不空则进行添加操作
            connectionManager.putChannel(ctx.channel());
        }
        // 启动心跳任务
        scheduledTask = GLOBAL_SCHEDULER.scheduleAtFixedRate(
                () -> sendHeartbeat(ctx.channel()),
                0, HEARTBEAT_INTERVAL_MS, TimeUnit.MILLISECONDS
        );
        //调用下一个handler的ChannelActive方法
        ctx.fireChannelActive();

    }

    /**
     * 重写 channelInactive 逻辑适配
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if (scheduledTask!=null){//讲该任务从 定时任务线程池中去除
            scheduledTask.cancel(false);
        }
        super.channelInactive(ctx);
    }
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, CustomMessage customMessage) throws Exception {
        if (customMessage.getType() == MessageType.HEARTBEAT_REQUEST) {
            handleHeartbeatRequest(channelHandlerContext, customMessage);
        } else if (customMessage.getType() == MessageType.HEARTBEAT_RESPONSE) {
            handleHeartbeatResponse(customMessage.getRequestId(), customMessage,channelHandlerContext.channel());
        } else {
            channelHandlerContext.fireChannelRead(customMessage);
        }
    }
    private void sendHeartbeat(Channel channel) {
        if (!channel.isActive()) return;

        long requestId = connectionManager.generateRequestId();
        CustomMessage heartbeat = new CustomMessage(
                ProtocolConstants.MAGIC,
                (byte) 1,
                (byte) 0x01,
                0,  // 保留字段
                MessageType.HEARTBEAT_REQUEST,
                requestId,
                0,
                new byte[0]
        );

        ExtendedChannelContext extendedChannelContext = connectionManager.getExtendedChannelContext(channel.id());
        //异步发送心跳包，设置超时时间为HEARTBEAT_TIMEOUT_MS 单位为 TimeUnit.MILLISECONDS
        extendedChannelContext.sendRequestAsync(requestId,heartbeat,HEARTBEAT_TIMEOUT_MS, TimeUnit.MILLISECONDS,
                (respMsg,ex)->{
                    if (ex != null) {
                        handleHeartbeatFailure(channel, ex);
                    } else {
                        timeoutCount.set(0);  // 重置连续超时计数
                        // 正常心跳响应处理
                        System.out.println("收到心跳包请求，正常做出响应: " + respMsg.getRequestId());
                    }
                });
    }
    private void handleHeartbeatFailure(Channel channel, Throwable ex) {
        if (ex instanceof TimeoutException) {
            int count = timeoutCount.incrementAndGet();
            System.err.println("心跳超时 (" + count + "/" + MAX_TIMEOUT_RETRY + ")");

            if (count >= MAX_TIMEOUT_RETRY) {
                System.err.println("由于连续超时失败而关闭渠道");
                channel.close();
            }
        } else {
            System.err.println("心跳失败: " + ex.getMessage());
            channel.close();
        }
    }
    private void handleHeartbeatRequest(ChannelHandlerContext ctx, CustomMessage request) {
        CustomMessage response = new CustomMessage(
                request.getMagic(),
                request.getVersion(),
                request.getSerializeAlgo(),
                0,  // 明确设置保留字段
                MessageType.HEARTBEAT_RESPONSE,
                request.getRequestId(),
                0,
                new byte[0]
        );
        ExtendedChannelContext extendedChannelContext = connectionManager.getExtendedChannelContext(ctx.channel().id());
        //发送响应包
        extendedChannelContext.sendResponse(response);
//        ctx.writeAndFlush(response);
    }


    private void handleHeartbeatResponse(long requestId, CustomMessage message,Channel channel) {
        ExtendedChannelContext channelContext = connectionManager.getExtendedChannelContext(channel.id());
        if (channelContext != null) {
/*            CompletableFuture<CustomMessage> future = channelContext.getPendingFuture(requestId);
            if (future != null && !future.isDone()) {
                future.complete(message);
            }*/
            channelContext.completeRequest(requestId,message);
        }
    }
}
